package com.shujia.streeam

import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo11DirectPartition {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[4]").setAppName("Demo8SSCOnKafka")

    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    val map = Map(
      "metadata.broker.list" -> "node2:9092,node3:9092,node4:9092",
      "auto.offset.reset" -> "smallest"
    )

    val topics = Set("student")

    /**
      * direct  模式 每一次job开始计算的时候才去broker中拉去数据
      * 偏移量不在存放到zk  而是需要自己管理
      * direct  rdd分区和kafka topic分区是一一对应的
      *
      *
      */
    val ds: InputDStream[(String, String)] = KafkaUtils.createDirectStream[
      String,
      String,
      StringDecoder,
      StringDecoder](
      ssc,
      map,
      topics
    )

    ds.foreachRDD(rdd => {

      //direct 模式  rdd分区和kafka分区一致
      println("rdd分区数：" + rdd.getNumPartitions)


      rdd.foreachPartition(iter => {
        println(iter.toList)
      })

      //通过每个年龄的人数
    })

    /**
      * shuffle  将相同的key拉去到同一个reduce中进行计算
      *
      * 由于生产者往kafka中生产数据使用的是hash的方法
      * 当通过spark streaming消费kafka数据使用direct模式
      * 每一个rdd中的分区对应kafka里面的一个分区  那么shuffle类的算子就不需要拉去数据了   速度更快
      *
      */

    ds
      .map(line => line._2.split(",")(2).toInt)
      .map((_, 1))
      .updateStateByKey((seq: Seq[Int], stat: Option[Int]) => Option(seq.sum + stat.getOrElse(0)))
      .print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
